package org.apache.rocketmq.broker.topic;

import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.BrokerPathConfigHelper;
import org.apache.rocketmq.common.*;
import org.apache.rocketmq.common.attribute.AttributeUtil;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.constant.PermName;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.protocol.DataVersion;
import org.apache.rocketmq.remoting.protocol.body.TopicConfigAndMappingSerializeWrapper;
import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingInfo;

import java.io.File;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import static com.google.common.base.Preconditions.checkNotNull;



/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/1/17
 * @方法描述：主题配置信息管理器，用于管理主题配置信息，包括主题的读写队列数量、权限、扩展属性等信息。
 */
public class TopicConfigManager extends ConfigManager {

    protected static final Logger log = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);


    //预定义的定时任务主题队列数
    private static final int SCHEDULE_TOPIC_QUEUE_NUM = 18;

    //存储主题和主题配置信息的Map，这个map中的主题和对应的配置信息都会注册到Namesrv中
    protected ConcurrentMap<String, TopicConfig> topicConfigTable = new ConcurrentHashMap<>(1024);

    //数据版本，这个数据版本的作用非常重要，每当主题配置信息发生变化时，都会更新数据版本，用于标识主题配置信息的变化
    private DataVersion dataVersion = new DataVersion();

    protected transient BrokerController brokerController;

    private transient final Lock topicConfigTableLock = new ReentrantLock();

    private static final long LOCK_TIMEOUT_MILLIS = 3000;


    //构造方法
    public TopicConfigManager() {
    }


    public TopicConfigManager(BrokerController brokerController) {
        this(brokerController, true);
    }


    public TopicConfigManager(BrokerController brokerController, boolean init) {
        this.brokerController = brokerController;
        //对主题配置信息进行初始化操作
        if (init) {
            init();
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/17
     * @方法描述：初始化主题配置信息的操作，这里初始化的主题配置信息都是系统内置的
     */
    protected void init() {
        {
            //得到系统自测主题名称
            String topic = TopicValidator.RMQ_SYS_SELF_TEST_TOPIC;
            //为该主题创建配置对象
            TopicConfig topicConfig = new TopicConfig(topic);
            //将该主题添加到系统主题集合中
            TopicValidator.addSystemTopic(topic);
            //设置读队列数量
            topicConfig.setReadQueueNums(1);
            //设置写队列数量
            topicConfig.setWriteQueueNums(1);
            //将主题配置添加到配置管理器中
            putTopicConfig(topicConfig);
        }

        //以下代码块中的主题都是系统内置主题，具体的操作和上面代码块中的内容一样，只不过就是主题的名称不同
        //所以我就不添加详细注释了，大家自己看看就行
        {
            //判断程序是否启用了当主题信息不存在时，是否自动创建主题的功能
            if (this.brokerController.getBrokerConfig().isAutoCreateTopicEnable()) {
                //获取自动创建主题的主题名称，这里我稍微解释一下，假如用户向"Test"主题中发送消息，但是这个主题不存在
                //这个时候并不是说要在Broker中创建TopicValidator.AUTO_CREATE_TOPIC_KEY_TOPIC这个主题，而是要创建Test这个主题
                //但是Test这个主题的配置信息，需要从TopicValidator.AUTO_CREATE_TOPIC_KEY_TOPIC这个主题中继承过来
                String topic = TopicValidator.AUTO_CREATE_TOPIC_KEY_TOPIC;
                TopicConfig topicConfig = new TopicConfig(topic);
                TopicValidator.addSystemTopic(topic);
                topicConfig.setReadQueueNums(this.brokerController.getBrokerConfig()
                        .getDefaultTopicQueueNums());
                topicConfig.setWriteQueueNums(this.brokerController.getBrokerConfig()
                        .getDefaultTopicQueueNums());
                //得到主题权限
                int perm = PermName.PERM_INHERIT | PermName.PERM_READ | PermName.PERM_WRITE;
                //设置主题权限
                topicConfig.setPerm(perm);
                putTopicConfig(topicConfig);
            }
        }

        {
            //得到程序性能基准测试主题，执行基准测试的时候就可以使用这个主题存储消息
            String topic = TopicValidator.RMQ_SYS_BENCHMARK_TOPIC;
            TopicConfig topicConfig = new TopicConfig(topic);
            TopicValidator.addSystemTopic(topic);
            topicConfig.setReadQueueNums(1024);
            topicConfig.setWriteQueueNums(1024);
            putTopicConfig(topicConfig);
        }


        {
            //得到当前Broker节点所在集群名称主题
            String topic = this.brokerController.getBrokerConfig().getBrokerClusterName();
            TopicConfig topicConfig = new TopicConfig(topic);
            TopicValidator.addSystemTopic(topic);
            int perm = PermName.PERM_INHERIT;
            if (this.brokerController.getBrokerConfig().isClusterTopicEnable()) {
                perm |= PermName.PERM_READ | PermName.PERM_WRITE;
            }
            topicConfig.setPerm(perm);
            putTopicConfig(topicConfig);
        }


        {
            //得到当前Broker节点名称主题
            String topic = this.brokerController.getBrokerConfig().getBrokerName();
            TopicConfig topicConfig = new TopicConfig(topic);
            TopicValidator.addSystemTopic(topic);
            int perm = PermName.PERM_INHERIT;
            if (this.brokerController.getBrokerConfig().isBrokerTopicEnable()) {
                perm |= PermName.PERM_READ | PermName.PERM_WRITE;
            }
            topicConfig.setReadQueueNums(1);
            topicConfig.setWriteQueueNums(1);
            topicConfig.setPerm(perm);
            putTopicConfig(topicConfig);
        }

        //以下主题在第七版本代码中都用不到，所以就先不添加注释了，其实上面的内置主题在第七版本代码中也用不到
        //我是添加注释到这里后感觉没必要这么早就给这些内容都加上注释，所以就先不写了，等后面用到的时候再为大家添加注释

        {
            String topic = TopicValidator.RMQ_SYS_OFFSET_MOVED_EVENT;
            TopicConfig topicConfig = new TopicConfig(topic);
            TopicValidator.addSystemTopic(topic);
            topicConfig.setReadQueueNums(1);
            topicConfig.setWriteQueueNums(1);
            putTopicConfig(topicConfig);
        }

        {

            String topic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC;
            TopicConfig topicConfig = new TopicConfig(topic);
            TopicValidator.addSystemTopic(topic);
            topicConfig.setReadQueueNums(SCHEDULE_TOPIC_QUEUE_NUM);
            topicConfig.setWriteQueueNums(SCHEDULE_TOPIC_QUEUE_NUM);
            putTopicConfig(topicConfig);
        }


        {

            if (this.brokerController.getBrokerConfig().isTraceTopicEnable()) {
                String topic = this.brokerController.getBrokerConfig().getMsgTraceTopicName();
                TopicConfig topicConfig = new TopicConfig(topic);
                TopicValidator.addSystemTopic(topic);
                topicConfig.setReadQueueNums(1);
                topicConfig.setWriteQueueNums(1);
                putTopicConfig(topicConfig);
            }
        }


        {
            String topic = this.brokerController.getBrokerConfig().getBrokerClusterName() + "_" + MixAll.REPLY_TOPIC_POSTFIX;
            TopicConfig topicConfig = new TopicConfig(topic);
            TopicValidator.addSystemTopic(topic);
            topicConfig.setReadQueueNums(1);
            topicConfig.setWriteQueueNums(1);
            putTopicConfig(topicConfig);
        }


        {
            String topic = PopAckConstants.buildClusterReviveTopic(this.brokerController.getBrokerConfig().getBrokerClusterName());
            TopicConfig topicConfig = new TopicConfig(topic);
            TopicValidator.addSystemTopic(topic);
            topicConfig.setReadQueueNums(this.brokerController.getBrokerConfig().getReviveQueueNum());
            topicConfig.setWriteQueueNums(this.brokerController.getBrokerConfig().getReviveQueueNum());
            putTopicConfig(topicConfig);
        }

        {
            String topic = TopicValidator.SYNC_BROKER_MEMBER_GROUP_PREFIX + this.brokerController.getBrokerConfig().getBrokerName();
            TopicConfig topicConfig = new TopicConfig(topic);
            TopicValidator.addSystemTopic(topic);
            topicConfig.setReadQueueNums(1);
            topicConfig.setWriteQueueNums(1);
            topicConfig.setPerm(PermName.PERM_INHERIT);
            putTopicConfig(topicConfig);
        }


        {

            String topic = TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC;
            TopicConfig topicConfig = new TopicConfig(topic);
            TopicValidator.addSystemTopic(topic);
            topicConfig.setReadQueueNums(1);
            topicConfig.setWriteQueueNums(1);
            putTopicConfig(topicConfig);
        }


        {
            String topic = TopicValidator.RMQ_SYS_TRANS_OP_HALF_TOPIC;
            TopicConfig topicConfig = new TopicConfig(topic);
            TopicValidator.addSystemTopic(topic);
            topicConfig.setReadQueueNums(1);
            topicConfig.setWriteQueueNums(1);
            putTopicConfig(topicConfig);
        }
    }



    protected TopicConfig putTopicConfig(TopicConfig topicConfig) {
        return this.topicConfigTable.put(topicConfig.getTopicName(), topicConfig);
    }

    protected TopicConfig getTopicConfig(String topicName) {
        return this.topicConfigTable.get(topicName);
    }

    protected TopicConfig removeTopicConfig(String topicName) {
        return this.topicConfigTable.remove(topicName);
    }

    public TopicConfig selectTopicConfig(final String topic) {
        return getTopicConfig(topic);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/17
     * @方法描述：更新主题配置信息的方法，该方法是本类中非常核心的方法，该方法虽然是更新主题配置信息操作，但假如主题配置信息map中并不存在方法参数中的主题，就会把主题配置信息添加到主题配置信息Map中
     */
    public void updateTopicConfig(final TopicConfig topicConfig) {
        //对要更新的主题配置信息进行非空校验
        checkNotNull(topicConfig, "topicConfig shouldn't be null");
        //得到主题配置信息的扩展属性
        Map<String, String> newAttributes = request(topicConfig);
        //得到当前存储的主题的扩展属性，也就是尚未更新的主题的扩展属性
        Map<String, String> currentAttributes = current(topicConfig.getTopicName());
        //对主题的扩展属性进行合并，这里的合并是指将新的扩展属性和当前的扩展属性合并到一起
        Map<String, String> finalAttributes = AttributeUtil.alterCurrentAttributes(
                this.topicConfigTable.get(topicConfig.getTopicName()) == null,
                TopicAttributes.ALL,
                ImmutableMap.copyOf(currentAttributes),
                ImmutableMap.copyOf(newAttributes));
        //把合并后的扩展属性设置到主题配置信息中
        topicConfig.setAttributes(finalAttributes);
        //将更新后的主题配置信息添加到主题配置信息Map中
        TopicConfig old = putTopicConfig(topicConfig);
        if (old != null) {//主题配置信息Map中原本就存在方法参数中的主题，那就会记录更新主题信息的日志
            log.info("update topic config, old:[{}] new:[{}]", old, topicConfig);
        } else {//如果主题配置信息Map中原本并不存在方法参数中的主题，那就会记录创建了新主题信息的日志
            log.info("create new topic [{}]", topicConfig);
        }
        //接下来有一个很重要的操作，那就是更新数据版本，因为主题配置信息发生了变化，所以数据版本也要更新一下
        long stateMachineVersion = brokerController.getMessageStore() != null ? brokerController.getMessageStore().getStateMachineVersion() : 0;
        dataVersion.nextVersion(stateMachineVersion);
        //每一次主题配置信息发生变化，都会将主题配置信息持久化到磁盘中
        //persist()方法是父类中的方法
        this.persist(topicConfig.getTopicName(), topicConfig);
    }


    //判断当前主题是否是顺序消息主题
    public boolean isOrderTopic(final String topic) {
        TopicConfig topicConfig = getTopicConfig(topic);
        if (topicConfig == null) {
            return false;
        } else {
            return topicConfig.isOrder();
        }
    }

    //移除指定主题的配置信息
    public void deleteTopicConfig(final String topic) {
        TopicConfig old = removeTopicConfig(topic);
        if (old != null) {
            log.info("delete topic config OK, topic: {}", old);
            long stateMachineVersion = brokerController.getMessageStore() != null ? brokerController.getMessageStore().getStateMachineVersion() : 0;
            dataVersion.nextVersion(stateMachineVersion);
            this.persist();
        } else {
            log.warn("delete topic config failed, topic: {} not exists", topic);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/17
     * @方法描述：创建主题配置信息序列化包装类对象的方法
     */
    public TopicConfigSerializeWrapper buildTopicConfigSerializeWrapper() {
        //创建一个主题配置信息序列化包装类对象
        TopicConfigSerializeWrapper topicConfigSerializeWrapper = new TopicConfigSerializeWrapper();
        //把主题配置信息Map设置到主题配置信息序列化包装类对象中
        topicConfigSerializeWrapper.setTopicConfigTable(this.topicConfigTable);
        DataVersion dataVersionCopy = new DataVersion();
        //得到当前主题配置信息的数据版本
        dataVersionCopy.assignNewOne(this.dataVersion);
        //把数据版本设置到主题配置信息序列化包装类对象中
        topicConfigSerializeWrapper.setDataVersion(dataVersionCopy);
        return topicConfigSerializeWrapper;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/17
     * @方法描述：为指定的主题配置信息创建序列化包装类对象的方法
     */
    public TopicConfigAndMappingSerializeWrapper buildSerializeWrapper(final ConcurrentMap<String, TopicConfig> topicConfigTable) {
        return buildSerializeWrapper(topicConfigTable, Maps.newHashMap());
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/17
     * @方法描述：为指定的主题配置信息和主题队列映射信息创建序列化包装类对象的方法，在第七版本代码中，这个主题队列映射信息还没有用到，也就是说该方法中的第二个参数是空的
     */
    public TopicConfigAndMappingSerializeWrapper buildSerializeWrapper(
            final ConcurrentMap<String, TopicConfig> topicConfigTable,
            final Map<String, TopicQueueMappingInfo> topicQueueMappingInfoMap) {
        //创建主题配置信息和主题队列映射信息序列化包装类对象
        TopicConfigAndMappingSerializeWrapper topicConfigWrapper = new TopicConfigAndMappingSerializeWrapper();
        topicConfigWrapper.setTopicConfigTable(topicConfigTable);
        topicConfigWrapper.setTopicQueueMappingInfoMap(topicQueueMappingInfoMap);
        topicConfigWrapper.setDataVersion(this.getDataVersion());
        if (this.brokerController.getBrokerConfig().isEnableSplitRegistration()) {
            this.getDataVersion().nextVersion();
        }
        return topicConfigWrapper;
    }



    //对主题配置信息进行编码的方法
    @Override
    public String encode() {
        return encode(false);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/17
     * @方法描述：得到主题配置信息的持久化文件路径的方法
     */
    @Override
    public String configFilePath() {
        return BrokerPathConfigHelper.getTopicConfigPath(this.brokerController.getMessageStoreConfig().getStorePathRootDir());
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/17
     * @方法描述：对主题配置信息进行解码的方法
     */
    @Override
    public void decode(String jsonString) {
        if (jsonString != null) {
            TopicConfigSerializeWrapper topicConfigSerializeWrapper =
                    TopicConfigSerializeWrapper.fromJson(jsonString, TopicConfigSerializeWrapper.class);
            if (topicConfigSerializeWrapper != null) {
                this.topicConfigTable.putAll(topicConfigSerializeWrapper.getTopicConfigTable());
                this.dataVersion.assignNewOne(topicConfigSerializeWrapper.getDataVersion());
                this.printLoadDataWhenFirstBoot(topicConfigSerializeWrapper);
            }
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/17
     * @方法描述：对主题配置信息进行编码的方法
     */
    public String encode(final boolean prettyFormat) {
        //创建一个主题配置信息序列化包装类对象
        TopicConfigSerializeWrapper topicConfigSerializeWrapper = new TopicConfigSerializeWrapper();
        topicConfigSerializeWrapper.setTopicConfigTable(this.topicConfigTable);
        topicConfigSerializeWrapper.setDataVersion(this.dataVersion);
        //对主题配置信息执行序列化操作
        return topicConfigSerializeWrapper.toJson(prettyFormat);
    }


    private void printLoadDataWhenFirstBoot(final TopicConfigSerializeWrapper tcs) {
        Iterator<Map.Entry<String, TopicConfig>> it = tcs.getTopicConfigTable().entrySet().iterator();
        while (it.hasNext()) {
            Map.Entry<String, TopicConfig> next = it.next();
            log.info("load exist local topic, {}", next.getValue().toString());
        }
    }

    public DataVersion getDataVersion() {
        return dataVersion;
    }

    public void setTopicConfigTable(
            ConcurrentMap<String, TopicConfig> topicConfigTable) {
        this.topicConfigTable = topicConfigTable;
    }

    public ConcurrentMap<String, TopicConfig> getTopicConfigTable() {
        return topicConfigTable;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/17
     * @方法描述：得到主题配置信息的扩展属性的方法
     */
    private Map<String, String> request(TopicConfig topicConfig) {
        return topicConfig.getAttributes() == null ? new HashMap<>() : topicConfig.getAttributes();
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/17
     * @方法描述：从主题配置信息map中得到当前存储的主题的扩展属性的方法
     */
    private Map<String, String> current(String topic) {
        //从主题配置信息map中得到当前存储的主题的配置信息
        TopicConfig topicConfig = getTopicConfig(topic);
        if (topicConfig == null) {
            return new HashMap<>();
        } else {//得到主题配置信息的扩展属性
            Map<String, String> attributes = topicConfig.getAttributes();
            if (attributes == null) {
                return new HashMap<>();
            } else {
                return attributes;
            }
        }
    }


    public boolean containsTopic(String topic) {
        return topicConfigTable.containsKey(topic);
    }


    public TopicConfig createTopicInSendMessageMethod(final String topic, final String defaultTopic,
                                                      final String remoteAddress, final int clientDefaultTopicQueueNums, final int topicSysFlag) {
        TopicConfig topicConfig = null;
        boolean createNew = false;

        try {
            if (this.topicConfigTableLock.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
                try {
                    topicConfig = getTopicConfig(topic);
                    if (topicConfig != null) {
                        return topicConfig;
                    }

                    TopicConfig defaultTopicConfig = getTopicConfig(defaultTopic);
                    if (defaultTopicConfig != null) {
                        if (defaultTopic.equals(TopicValidator.AUTO_CREATE_TOPIC_KEY_TOPIC)) {
                            if (!this.brokerController.getBrokerConfig().isAutoCreateTopicEnable()) {
                                defaultTopicConfig.setPerm(PermName.PERM_READ | PermName.PERM_WRITE);
                            }
                        }

                        if (PermName.isInherited(defaultTopicConfig.getPerm())) {
                            topicConfig = new TopicConfig(topic);

                            int queueNums = Math.min(clientDefaultTopicQueueNums, defaultTopicConfig.getWriteQueueNums());

                            if (queueNums < 0) {
                                queueNums = 0;
                            }

                            topicConfig.setReadQueueNums(queueNums);
                            topicConfig.setWriteQueueNums(queueNums);
                            int perm = defaultTopicConfig.getPerm();
                            perm &= ~PermName.PERM_INHERIT;
                            topicConfig.setPerm(perm);
                            topicConfig.setTopicSysFlag(topicSysFlag);
                            topicConfig.setTopicFilterType(defaultTopicConfig.getTopicFilterType());
                        } else {
                            log.warn("Create new topic failed, because the default topic[{}] has no perm [{}] producer:[{}]",
                                    defaultTopic, defaultTopicConfig.getPerm(), remoteAddress);
                        }
                    } else {
                        log.warn("Create new topic failed, because the default topic[{}] not exist. producer:[{}]",
                                defaultTopic, remoteAddress);
                    }

                    if (topicConfig != null) {
                        log.info("Create new topic by default topic:[{}] config:[{}] producer:[{}]",
                                defaultTopic, topicConfig, remoteAddress);

                        putTopicConfig(topicConfig);

                        long stateMachineVersion = brokerController.getMessageStore() != null ? brokerController.getMessageStore().getStateMachineVersion() : 0;
                        dataVersion.nextVersion(stateMachineVersion);

                        createNew = true;

                        this.persist();
                    }
                } finally {
                    this.topicConfigTableLock.unlock();
                }
            }
        } catch (InterruptedException e) {
            log.error("createTopicInSendMessageMethod exception", e);
        }

        if (createNew) {
            registerBrokerData(topicConfig);
        }

        return topicConfig;
    }


    public TopicConfig createTopicInSendMessageBackMethod(
            final String topic,
            final int clientDefaultTopicQueueNums,
            final int perm,
            final int topicSysFlag) {
        return createTopicInSendMessageBackMethod(topic, clientDefaultTopicQueueNums, perm, false, topicSysFlag);
    }

    public TopicConfig createTopicInSendMessageBackMethod(
            final String topic,
            final int clientDefaultTopicQueueNums,
            final int perm,
            final boolean isOrder,
            final int topicSysFlag) {
        TopicConfig topicConfig = getTopicConfig(topic);
        if (topicConfig != null) {
            if (isOrder != topicConfig.isOrder()) {
                topicConfig.setOrder(isOrder);
                this.updateTopicConfig(topicConfig);
            }
            return topicConfig;
        }

        boolean createNew = false;

        try {
            if (this.topicConfigTableLock.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
                try {
                    topicConfig = getTopicConfig(topic);
                    if (topicConfig != null) {
                        return topicConfig;
                    }

                    topicConfig = new TopicConfig(topic);
                    topicConfig.setReadQueueNums(clientDefaultTopicQueueNums);
                    topicConfig.setWriteQueueNums(clientDefaultTopicQueueNums);
                    topicConfig.setPerm(perm);
                    topicConfig.setTopicSysFlag(topicSysFlag);
                    topicConfig.setOrder(isOrder);

                    log.info("create new topic {}", topicConfig);
                    putTopicConfig(topicConfig);
                    createNew = true;
                    long stateMachineVersion = brokerController.getMessageStore() != null ? brokerController.getMessageStore().getStateMachineVersion() : 0;
                    dataVersion.nextVersion(stateMachineVersion);
                    this.persist();
                } finally {
                    this.topicConfigTableLock.unlock();
                }
            }
        } catch (InterruptedException e) {
            log.error("createTopicInSendMessageBackMethod exception", e);
        }

        if (createNew) {
            registerBrokerData(topicConfig);
        }

        return topicConfig;
    }

    private void registerBrokerData(TopicConfig topicConfig) {
        if (brokerController.getBrokerConfig().isEnableSingleTopicRegister()) {
            this.brokerController.registerSingleTopicAll(topicConfig);
        } else {
            this.brokerController.registerIncrementBrokerData(topicConfig, dataVersion);
        }
    }
}
